package algebra.visitors.cursors;

import java.util.HashSet;
import java.util.Set;

import algebra.op.AlgebraicOP;
import algebra.op.JoinOP;
import entities.Individual;
import entities.assertion.ConceptAssertion;
import entities.concept.Concept;

public class CACursorJoin extends CACursor {
	public static enum JoinAlgorithm {
		NESTED_LOOP, MERGE_JOIN
	};

	// default: nested loop join
	private JoinAlgorithm join = JoinAlgorithm.NESTED_LOOP;
	private final AlgebraicOP sub1;
	private final AlgebraicOP sub2;
	private ConceptAssertion ca1, ca2;
	private CACursor c1, c2;
	private boolean ls, rs;

	public JoinAlgorithm getJoinAlg(){
		return join;
	}
	// private boolean emptyRel = false;

	public CACursorJoin(JoinOP ip) {
		this.sub1 = ip.getFirstSubQuery();
		this.sub2 = ip.getSecondSubQuery();
		ca1 = null;
		ca2 = null;
	}

//	public void setJoinAlgorithm(JoinAlgorithm type) {
//		join = type;
//	}

	@Override
	public boolean openCursor() {
		// determine the type of algorithm
		c1 = CACursor.getNewCACursor(sub1);
		c2 = CACursor.getNewCACursor(sub2);
		ls = c1 instanceof CACursorConstant;
		rs = c2 instanceof CACursorConstant;
		c1.openCursor();
		c2.openCursor();
		//determine join algorithm
		join = chooseJoin(sub1, sub2);
		return false;
	}

	public static JoinAlgorithm chooseJoin(AlgebraicOP left, AlgebraicOP right) {
		return chooseJoin(CACursor.getNewCACursor(left), CACursor.getNewCACursor(right));
	}
	
	private static JoinAlgorithm chooseJoin(CACursor lc, CACursor rc) {
		lc.openCursor();
		rc.openCursor();
		if(lc.isSortedById() && rc.isSortedById()){
			return JoinAlgorithm.MERGE_JOIN;
		}
		return JoinAlgorithm.NESTED_LOOP;
	}
	@Override
	public ConceptAssertion getNextResult() {
		if (ls && rs) {
			return this.getNextResultTwoStar();
		} else if (ls) {
			// one star: c1
			return this.getNextResultOneStar(c1, c2);
		} else if (rs) {
			return this.getNextResultOneStar(c2, c1);
		} else {
			ca1 = c1.getNextResult();
			// no star
			if (join == JoinAlgorithm.NESTED_LOOP) {
				return zeroStarNestedLoopJoin();
			} else if (join == JoinAlgorithm.MERGE_JOIN) {
				return this.zeroStarMergeJoin();
			} else {
				System.out.println("Not implemnted yet for join: "
						+ join.name());
			}
		}
		return null;
	}

	/**
	 * We assume the left and right args were already sorted by ind names
	 * 
	 * @return
	 */
	private ConceptAssertion zeroStarMergeJoin() {
		if (ca1 == null) {
			return null;
		}
		ca2 = c2.getNextResult();
		while (ca1 != null && ca2 != null) {
			int comp = ca1.getIndividual().compareTo(ca2.getIndividual());
			if (comp < 0) {
				// advance cursor 1
				ca1 = c1.getNextResult();
				continue;
			} else if (comp > 0) {
				// advance cursor 2
				ca2 = c2.getNextResult();
				continue;
			} else {
				// join
				Set<Concept> cd = new HashSet<Concept>();
				cd.add(ca1.getConcept());
				cd.add(ca2.getConcept());
				ConceptAssertion ret = new ConceptAssertion(
						ca1.getIndividual(), Concept.getConjunctiveConcept(cd));
				return ret;
			}
		}
		return null;
	}

	/**
	 * Note that we assume sub1 has smaller size for standard semantics. Also:
	 * the algorithm below assumes no empty set for either sub-cursors.
	 * 
	 * @author j55wu
	 * 
	 */
	private ConceptAssertion zeroStarNestedLoopJoin() {
		if (ca1 == null) {
			return null;
		}
		c2 = CACursor.getNewCACursor(sub2);
		c2.openCursor();
		while(ca1  != null ){
			ca2 = c2.getNextResult();
			while(ca2 != null){
				if (ca1.getIndividual().equals(ca2.getIndividual())) {
					Set<Concept> cd = new HashSet<Concept>();
					cd.add(ca1.getConcept());
					cd.add(ca2.getConcept());
					ConceptAssertion ret = new ConceptAssertion(ca1.getIndividual(),
							Concept.getConjunctiveConcept(cd));
					return ret;
				}else{
					ca2 = c2.getNextResult();
				}
			}
			//find another ca1
			ca1 = c1.getNextResult();
			// c1 also exhausted
			if (ca1 == null) {
				break;
			}
			c2 = CACursor.getNewCACursor(sub2);
			c2.openCursor();
		}
		return null;
	}
	
	
//	private ConceptAssertion zeroStarNestedLoopJoin() {
//		if (ca1 == null) {
//			return null;
//		}
//		ca2 = c2.getNextResult();
//		if (ca2 == null) {
//			// all c2 exhausted for c1
//			ca1 = c1.getNextResult();
//			// c1 also exhausted
//			if (ca1 == null) {
//				return null;
//			}
//			// restart c2
//			c2 = CACursor.getNewCACursor(sub2);
//			c2.openCursor();
//			ca2 = c2.getNextResult();
//			if(ca2 == null){
//				return null;
//			}
//		}
//		if (ca1.getIndividual().equals(ca2.getIndividual())) {
//			Set<Concept> cd = new HashSet<Concept>();
//			cd.add(ca1.getConcept());
//			cd.add(ca2.getConcept());
//			return new ConceptAssertion(ca1.getIndividual(),
//					Concept.getConjunctiveConcept(cd));
//		} else {
//			// no match, try next ca2
////			System.out.println("ca1: "+ca1);
////			System.out.println("ca2: "+ca2);
//			return zeroStarNestedLoopJoin();
//		}
//	}

	private Concept cp;

	private ConceptAssertion getNextResultOneStar(CACursor cs, CACursor ns) {
		if (cp == null) {
			cp = cs.getNextResult().getConcept();
		}
		ConceptAssertion sub = ns.getNextResult();
		if (sub != null) {
			Set<Concept> cd = new HashSet<Concept>();
			cd.add(sub.getConcept());
			cd.add(cp);
			return new ConceptAssertion(sub.getIndividual(),
					Concept.getConjunctiveConcept(cd));
		}
		return null;
	}

	private int count = 0;

	private ConceptAssertion getNextResultTwoStar() {
		if (count > 0) {
			return null;
		}
		count++;
		Set<Concept> cd = new HashSet<Concept>();
		cd.add(c1.getNextResult().getConcept());
		cd.add(c2.getNextResult().getConcept());
		return new ConceptAssertion(Individual.getStarIndividual(),
				Concept.getConjunctiveConcept(cd));
	}

	@Override
	public boolean isSortedById() {
		//if sort-merge is used
		//or: nested loops and left/outer subcursor is sorted
		return join.equals(JoinAlgorithm.MERGE_JOIN) || this.c1.isSortedById();
	}

}
